
#include "workStealing.h"

extern	vector<Node*>	nodes_ready;	// Nodes current machine can work on
extern	vector<Node*>	nodes_blocked; //  Nodes current machine is waiting for
extern	vector<Node*>	nodes;
extern	int				machine_id;	// Our machine id
extern	int				nb_machines;
extern	bool			no_more_task_global;

void print_MPI_error(int error_code)
{
	if(error_code!=MPI_SUCCESS) {
		switch(error_code) {
			case MPI_ERR_COMM: cout << time(NULL)<<" " << "ERROR: MPI_ERR_COMM" << endl; break;
			case MPI_ERR_COUNT: cout << time(NULL)<<" " << "ERROR: MPI_ERR_COUNT  " << endl; break;
			case MPI_ERR_TYPE  : cout << time(NULL)<<" " << "ERROR: MPI_ERR_TYPE  " << endl; break;
			case MPI_ERR_TAG  : cout << time(NULL)<<" " << "ERROR: MPI_ERR_TAG  " << endl; break;
			case MPI_ERR_RANK  : cout << time(NULL)<<" " << "ERROR: MPI_ERR_RANK  " << endl; break;
			default: cout << time(NULL)<<" " << "ERROR in mpi : unknown" << endl;
		}
	}
}


void * communication_dispatcher_function(void *arg)
{
/*	while(!all_tasks_ended) {
		char request_tag;
		MPI_Status stat;
		cout << time(NULL) << " " << machine_id << " Ready to listen..." << endl;
		//print_MPI_error(MPI_Recv(&request_tag, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &stat));
		MPI_Request request;
		print_MPI_error(MPI_Irecv(&request_tag, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &request));
		int flag_message_received = false;
		while(!flag_message_received) {
			//MPI_Wait(&request, &stat);
			//if(machine_id==1) 
			//cout << time(NULL) << " " << machine_id << " listening..." << endl;
			MPI_Test(&request, &flag_message_received, &stat);
			if(!flag_message_received) {
				usleep(1000);
			}
		}
		cout << time(NULL)<<" " << machine_id << " Received message from " << stat.MPI_SOURCE;
		if(stat.MPI_SOURCE!=machine_id) {
			switch(request_tag) {
				case TAG_WORKSTEAL_REQUEST:
					cout << " TAG_WORKSTEAL_REQUEST" << endl;
					work_stealing_undergone_thread_function(stat.MPI_SOURCE);
					break;
				case TAG_NODE_END:
					cout << " TAG_NODE_END" << endl;
					task_end_thread_function(stat.MPI_SOURCE);
					break;
				case TAG_LAST_TASK_END:
					cout << " TAG_LAST_TASK_END" << endl;
					last_node_thread_function(stat.MPI_SOURCE);
					break;
			}
		}
	}*/
	return NULL;
}

// Work stealing
void work_stealing()
{
	bool work_stealing_success = false;
	bool no_more_task	= true;
	//while(!work_stealing_success && !all_tasks_ended && nodes_ready.empty())
	{
		for(int i=0; i<nb_machines && !all_tasks_ended; i++) {
			int machine_to_steal_id = (i+machine_id)%nb_machines;
			if(machine_to_steal_id!=machine_id) {
				char request = TAG_WORKSTEAL_REQUEST;
				print_MPI_error(MPI_Send(&request, 1, MPI_CHAR, machine_to_steal_id, TAG_REQUEST, MPI_COMM_WORLD));
				print_MPI_error(MPI_Send(&machine_id, 1, MPI_INT, machine_to_steal_id, TAG_WORKSTEAL_REQUEST, MPI_COMM_WORLD));
#ifdef DEBUG
				cout << time(NULL)<<" " << machine_id << " trying to steal " << machine_to_steal_id << endl;
#endif
				int node_stolen_id;
				int mpi_ret;
				//print_MPI_error(mpi_ret=MPI_Recv(&node_stolen_id, 1, MPI_INT, machine_to_steal_id, TAG_WORKSTEAL_ANSWER, MPI_COMM_WORLD, NULL));
				// Using Irecv, to prevent deadlock at the end
				MPI_Request request_mpi;
				MPI_Status  stat;
				print_MPI_error(mpi_ret=MPI_Irecv(&node_stolen_id, 1, MPI_INT, machine_to_steal_id, TAG_WORKSTEAL_ANSWER, MPI_COMM_WORLD, &request_mpi));
				int flag_message_received = false;
				while(!flag_message_received) {
					MPI_Test(&request_mpi, &flag_message_received, &stat);
					if(!flag_message_received) {
						usleep(100);
					}
					if(all_tasks_ended) {
						return;
					}
				}
				
				if(mpi_ret==MPI_SUCCESS && node_stolen_id>=0) { // WORKSTEAL SUCCESS
#ifdef DEBUG
					cout << time(NULL)<<" " << machine_id << " WORKSTEAL SUCCESS on " << machine_to_steal_id << "(task " << node_stolen_id << " stolen)" <<endl;
#endif
					// prevent all nodes we are the new node's owner: TODO later, if needed (communication overhead)
					/*Node *node_stolen	= nodes[node_stolen_id];
					for(list<Node*>::iterator it=node_stolen->listDependant.begin();it!=node_stolen->listDependant.end(); ++it) {
						Node *node_to_prevent = *it;
						print_MPI_error(MPI_Send(&node_stolen_id, 1, MPI_INT, node_to_prevent->id_machine, TAG_UPDATE_NODE, MPI_COMM_WORLD));
					}*/
					work_stealing_success = true;
					no_more_task	= false;
					Node *node_steal = nodes[node_stolen_id];
					pthread_mutex_lock(&nodes_ready_mutex);
					nodes_ready.push_back(node_steal);
					pthread_mutex_unlock(&nodes_ready_mutex);
					break;
				} else {
#ifdef DEBUG
					cout << time(NULL)<<" " << machine_id << " WORKSTEAL FAILED on " << machine_to_steal_id << endl;
#endif
					if(node_stolen_id==-2) {
						no_more_task = no_more_task && true;
					} else {
						no_more_task = false; // no_more_task && false
					}
				}
			}
		}
		//sleep(1);
		if(no_more_task && nodes_ready.empty() && nodes_blocked.empty()) {
			all_tasks_ended = true;
		}
	}
}

void *work_stealing_undergone_thread_function(void *arg)
{
#ifdef DEBUG
	cout << time(NULL)<<" " << "work_stealing_undergone_thread launched" << endl;
#endif
	while(!all_tasks_ended) {
		// MPI receive... (waiting for someone trying to steal us)
		int machine_stealer_id, node_stolen_id = -1;
		Node*	node_stolen;
		MPI_Request request;
		MPI_Status  stat;
		print_MPI_error(MPI_Irecv(&machine_stealer_id, 1, MPI_INT, MPI_ANY_SOURCE, TAG_WORKSTEAL_REQUEST, MPI_COMM_WORLD, &request));
		int flag_message_received = false;
		while(!flag_message_received) {
			MPI_Test(&request, &flag_message_received, &stat);
			if(!flag_message_received) {
				usleep(1000);
			}
		}
#ifdef DEBUG
		cout << time(NULL)<<" " << machine_id << " : " << machine_stealer_id << " trying to steal us" << endl;
#endif
		// if nodes_ready.empty() : no nodes to give, answer "workstealing failed"
		pthread_mutex_lock(&nodes_ready_mutex);
		if(nodes_ready.empty()) {
#ifdef DEBUG
			cout << time(NULL) << " " << machine_id << "we have no nodes ready" << endl;
#endif
			if(nodes_blocked.empty()) { // We have no more nodes at all
#ifdef DEBUG
				cout << time(NULL) << " " << machine_id << "we have no nodes at all" << endl;
#endif
				node_stolen_id=-2;
				print_MPI_error(MPI_Send(&node_stolen_id, 1, MPI_INT, machine_stealer_id, TAG_WORKSTEAL_ANSWER, MPI_COMM_WORLD));
				node_stolen_id=-1;
			} else {
				print_MPI_error(MPI_Send(&node_stolen_id, 1, MPI_INT, machine_stealer_id, TAG_WORKSTEAL_ANSWER, MPI_COMM_WORLD));
			}
		} else {
			// trying to find a node which stealer is dependant
			//pthread_mutex_lock(&nodes_ready_mutex);
			//cout << time(NULL) << " " << machine_id << "DEBUG 0" << endl;
			for(vector<Node*>::iterator it_search=nodes_ready.begin();it_search!=nodes_ready.end() && -1==node_stolen_id;++it_search) {
				Node *node_ready = *it_search;
				//cout << time(NULL) << " " << machine_id << "DEBUG 0"<< *node_ready << endl;
				for(list<Node*>::iterator it_dep=node_ready->listDependant.begin();
						it_dep!=node_ready->listDependant.end() && -1==node_stolen_id; ++it_dep) {
					Node *node_dep = *it_dep;
					//cout << time(NULL) << " " << machine_id << "DEBUG 1"<< *node_dep << endl;
					if(node_dep->id_machine==machine_stealer_id) {
						//cout << time(NULL) << " " << machine_id << "DEBUG 2" << endl;
						node_stolen	= node_ready;
						//cout << time(NULL) << " " << machine_id << "DEBUG 2" << endl;
						nodes_ready.erase(it_search);
						//cout << time(NULL) << " " << machine_id << "DEBUG 2" << endl;
						node_stolen_id = node_ready->id;
						//cout << time(NULL) << " " << machine_id << "DEBUG 2" << endl;
						break;
					}
				}
			}
			//pthread_mutex_unlock(&nodes_ready_mutex);
			if(node_stolen_id==-1) { // if not found, give a nodes_ready (doesn't matter what node)
				//pthread_mutex_lock(&nodes_ready_mutex);
				node_stolen	= nodes_ready.back();
				// update nodes_ready
				nodes_ready.pop_back();
				node_stolen_id = node_stolen->id;
			}
			// send node_id (workStealing success)
#ifdef DEBUG
			cout << time(NULL) << " " << machine_id << " Sending node "<< node_stolen_id<<" to "<< machine_stealer_id << endl;
#endif
			print_MPI_error(MPI_Send(&node_stolen_id, 1, MPI_INT, machine_stealer_id, TAG_WORKSTEAL_ANSWER, MPI_COMM_WORLD));
		}
		pthread_mutex_unlock(&nodes_ready_mutex);
	}
}

// Node update
void update_node(Node *node_to_prevent, int node_id, int new_machine_owner_id)
{
}

void update_node_thread_function(int machine_source_id)
{
}

// Prevent tasks end, for dependencies
void task_end(Node *node_to_prevent, int node_ended_id)
{
}

void *task_end_thread_function(void *arg)
{
	cout << time(NULL)<<" " << machine_id << " task_end_thread_function LAUNCHED" << endl;
	while(!all_tasks_ended) {
		int node_ended_id;
		//print_MPI_error(MPI_Recv(&node_ended_id, 1, MPI_INT, MPI_ANY_SOURCE, TAG_NODE_END, MPI_COMM_WORLD, NULL));
		MPI_Request request;
		MPI_Status  stat;
		//print_MPI_error(MPI_Irecv(&tmp, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_LAST_TASK_END, MPI_COMM_WORLD, &request));
		print_MPI_error(MPI_Irecv(&node_ended_id, 1, MPI_INT, MPI_ANY_SOURCE, TAG_NODE_END, MPI_COMM_WORLD, &request));
		int flag_message_received = false;
		while(!flag_message_received) {
			MPI_Test(&request, &flag_message_received, &stat);
			if(!flag_message_received) {
				usleep(1000);
			}
		}
		cout << time(NULL)<<" " << machine_id << " Node " << node_ended_id << " ended" << endl;
		Node *node = nodes[node_ended_id];
		/*if(node_ended_id==root_node_id) {
		  all_tasks_ended = true;
		  cout << time(NULL)<<" " << machine_id << "COMPUTATION ENDED" << endl;
		  return;
		  }*/
		for(list<Node*>::iterator it=node->listDependant.begin();it!=node->listDependant.end();++it) {
			Node *node_dependant = *it;
			if(node_dependant->id_machine==machine_id) {
				task_ended_message_received(node_dependant->id, node->id);
			}
		}
	}
	cout << time(NULL)<<" " << machine_id << " task_end_thread_function END" << endl;
}

void *last_node_thread_function(void *arg)
{
	while(!all_tasks_ended) {
		cout << time(NULL)<<" " << machine_id << " last_node_thread_function LAUNCHED" << endl;
		char tmp;
		//print_MPI_error(MPI_Recv(&tmp, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_LAST_TASK_END, MPI_COMM_WORLD, NULL));
		MPI_Request request;
		MPI_Status  stat;
		//print_MPI_error(MPI_Irecv(&request_tag, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_REQUEST, MPI_COMM_WORLD, &request));
		print_MPI_error(MPI_Irecv(&tmp, 1, MPI_CHAR, MPI_ANY_SOURCE, TAG_LAST_TASK_END, MPI_COMM_WORLD, &request));
		int flag_message_received = false;
		while(!flag_message_received) {
			//MPI_Wait(&request, &stat);
			//if(machine_id==1) 
			//cout << time(NULL) << " " << machine_id << " listening..." << endl;
			MPI_Test(&request, &flag_message_received, &stat);
			if(!flag_message_received) {
				usleep(1000);
			}
		}
		cout << time(NULL)<<" " << machine_id << "Last task is gonna be executed: update boolean" << endl;
		//if(tmp==1) {
		all_tasks_ended=true;
		//}
	}
}
